package cn.gwm.flink.streaming.function.map;

import cn.gwm.flink.streaming.client.RedisClient;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.Map;

/**
 * @author GW00256253
 */
public class GpsToRedisFunction implements FlatMapFunction<JSONObject,JSONObject> {

    private String REDIS_KEY_LAST_GPS = "l_gps:";
    private int SECONDS = 259200;

    private String vehicleType;

    public GpsToRedisFunction(String vehicleType){
        this.vehicleType = vehicleType;
    }

    @Override
    public void flatMap(JSONObject entries, Collector<JSONObject> collector) throws Exception {
        JSONObject object = JSONUtil.createObj();
        object.set(BaseFields.tid,entries.getStr(BaseFields.tid));
        object.set("longitude", entries.getStr("tboxMoniLongitude"));
        object.set("latitude", entries.getStr("tboxMoniLatitude"));
        object.set(BaseFields.vin, entries.getStr(BaseFields.vin));
        collector.collect(object);
        RedisClient.setData(getKey(entries),object.toString(),SECONDS);
    }

    private String getKey(JSONObject entries){
        return REDIS_KEY_LAST_GPS+vehicleType+":"+entries.getStr(BaseFields.vin);
    }

}
